package com.hdu.dwh.dao;


import com.hdu.dwh.enums.HttpStatusEnum;
import com.hdu.dwh.exception.CustomException;
import com.hdu.dwh.utils.StatementUtils;

import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.UniqueConstraint;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.stream.Collectors;

@Slf4j
@Component
@RequiredArgsConstructor
@Data
public class FlinkTableDao {
    private final TableEnvironment tableEnvironment;
    private final FlinkBaseDao flinkBaseDao;


    /**
     * 删除表 true成功 无false 失败抛异常
     * @param catalog
     * @param database
     * @param table
     * @return
     */
    public boolean deleteTable(String catalog, String database, String table) {
        try{
            if(!flinkBaseDao.checkTableExists(catalog, database, table)){
                throw new CustomException(HttpStatusEnum.TABLE_NOT_EXISTS);
            }
            tableEnvironment.executeSql(String.format("DROP TABLE %s.%s.%s", catalog, database, table));
            return true;
        } catch (Exception e){
            throw new CustomException(HttpStatusEnum.FLINK_ERROR);
        }
    }

    /**
     * 获取建表 SQL 需确保table存在
     * @param table
     * @param databaseName
     * @param tableName
     * @return 生成的建表 SQL 字符串
     */
    public String generateCreateTableSQL(Table table, String catalog, String databaseName, String tableName) {
        ResolvedSchema schema = table.getResolvedSchema();
        List<String> columnDefs = schema.getColumns().stream()
                .map(column -> column.getName() + " " + column.getDataType())
                .collect(Collectors.toList());

        String primaryKey = "";
        if (schema.getPrimaryKey().isPresent()) {
            UniqueConstraint pkConstraint = schema.getPrimaryKey().get();
            List<String> pkColumns = pkConstraint.getColumns();
            primaryKey = ",\n  PRIMARY KEY (" + String.join(", ", pkColumns) + ") NOT ENFORCED";
        }

        String columnPart = String.join(",\n  ", columnDefs);

        return String.format("CREATE TABLE %s.%s.%s (\n  %s%s\n);",
                catalog, databaseName, tableName, columnPart, primaryKey);
    }


    /**
     * 创建表 true就是创建成功 false就是表已经存在
     * @param catalog
     * @param database
     * @param table
     * @param createSql
     * @return
     */
    public boolean createTable(String catalog, String database, String table, String createSql) {
        try{
            if(!flinkBaseDao.checkCatalogExists(catalog)){
                throw new CustomException(HttpStatusEnum.CATALOG_NOT_EXISTS);
            }
            if(!flinkBaseDao.checkDatabaseExists(catalog, database)){
                // 新建数据库
                flinkBaseDao.createDatabase(catalog, database);
                // 新建表
                tableEnvironment.executeSql(createSql);
                return true;
            } else{
                // 直接往里插 要判断表是否存在
                if(!flinkBaseDao.checkTableExists(catalog, database, table)){
                    // 表不存在 直接建表
                    tableEnvironment.executeSql(createSql);
                    return true;
                } else{
                    return false;
                }
            }
        } catch (Exception e){
            throw new CustomException(HttpStatusEnum.FLINK_ERROR);
        }
    }


    /**
     * 插入表值 true成功 无false 失败抛异常
     * @param table
     * @param catalogName
     * @param databaseName
     * @param tableName
     * @return
     */
    public boolean insertTable(Table table, String catalogName, String databaseName, String tableName){
        try{
            if(!flinkBaseDao.checkTableExists(catalogName, databaseName, tableName)){
                throw new CustomException(HttpStatusEnum.TABLE_NOT_EXISTS);
            }
            System.out.println("insert :"+tableName);
            table.executeInsert(String.format("%s.%s.%s", catalogName, databaseName, tableName));
            return true;
        } catch (Exception e){
            throw new CustomException(HttpStatusEnum.FLINK_ERROR);
        }
    }

    /**
     * 复制表 sinkTable如果已经存在 就不会复制了 返回true 成功 无 false 失败抛异常
     * @param sourceCatalog
     * @param sinkCatalog
     * @param sourceDatabase
     * @param sinkDatabase
     * @param sourceTable
     * @param sinkTable
     * @return
     */
    public boolean copyTable(String sourceCatalog, String sinkCatalog, String sourceDatabase,String sinkDatabase, String sourceTable, String sinkTable){
        try{
            if(!flinkBaseDao.checkTableExists(sourceCatalog, sourceDatabase, sourceTable)){
                throw new CustomException(HttpStatusEnum.TABLE_NOT_EXISTS);
            }
            Table table = tableEnvironment.from(String.format("%s.%s.%s", sourceCatalog, StatementUtils.quote(sourceDatabase), sourceTable));
            String createSql = generateCreateTableSQL(table, sinkCatalog, sinkDatabase, sinkTable);
            System.out.println(createSql);
            if(createTable(sinkCatalog, sinkDatabase, sinkTable, createSql)){
                insertTable(table, sinkCatalog, sinkDatabase, sinkTable);
            }
            return true;
        } catch (Exception e){
            e.printStackTrace();
            throw new CustomException(HttpStatusEnum.FLINK_ERROR);
        }
    }


    /**
     * 复制所有表 返回 true 成功 无false 失败抛异常
     * @param sourceCatalog
     * @param sinkCatalog
     * @param sourceDatabase
     * @param sinkDatabase
     * @return
     */
    public boolean copyAllTable(String sourceCatalog, String sinkCatalog, String sourceDatabase, String sinkDatabase){
        try{
            if(!flinkBaseDao.checkDatabaseExists(sourceCatalog, sourceDatabase)){
                throw new CustomException(HttpStatusEnum.DATABASE_NOT_EXISTS);
            }
            String[] tables = flinkBaseDao.showTable(sourceCatalog, sourceDatabase);
            for (String table : tables) {
                // table 不存在 就 复制表
                System.out.println(table);
                copyTable(sourceCatalog, sinkCatalog, sourceDatabase, sinkDatabase, table, table);
            }
            return true;
        } catch (Exception e){
            throw new CustomException(HttpStatusEnum.FLINK_ERROR);
        }
    }
    // /**
    //  * DIM层表创建
    //  * @param databaseA
    //  * @param tableA
    //  * @param databaseB
    //  * @param tableB
    //  */
    // public void createDimTable(String databaseA, String tableA, String databaseB, String tableB) {
    //     MySqlSource<String> source = MySqlSource.<String>builder()
    // .hostname("your-host")
    // .port(3306)
    // .databaseList("your_db")
    // .tableList("your_db.your_table")
    // .username("xxx")
    // .password("xxx")
    // .deserializer(new JsonDebeziumDeserializationSchema())
    // .startupOptions(StartupOptions.initial())  // 初始+增量
    // .build();

    // // 加这个：任务在 snapshot + binlog 消费完成后自动退出（只适用于测试或定期采集）
    // flinkBaseDao.getTableEnvironment().fromSource(source, WatermarkStrategy.noWatermarks(), "MySQL CDC").executeAndCollect() // 可以控制输出后退出

    // }



}
